package org.apache.spark.day02

import org.apache.spark.rdd.{MapPartitionsRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 对RDD进行操作，本质上是对RDD里面的每一个分区进行操作，一个分区其实就是对应一个迭代器
 * 对RDD进行map操作，就是对RDD里面的每一个分区对应的迭代器进行map操作
 */
object MapPartitionsRddDemo {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("data").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val array = Array(1,2,3, 4,5,6, 7,8,9,10)
    val dataRDD: RDD[Int] = sc.parallelize(array, 3)

    val mapRDD: RDD[Int] = dataRDD.map(_ * 10)

    /**
     *    调用迭代器的map方法，数据是一个一个的来
     *   def map[U: ClassTag](f: T => U): RDD[U] = withScope {
            val cleanF = sc.clean(f)
            new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
          }
            将整个迭代器传入到MapPartitionsRDD
            def mapPartitions[U: ClassTag](
                f: Iterator[T] => Iterator[U],
                preservesPartitioning: Boolean = false): RDD[U] = withScope {
              val cleanedF = sc.clean(f)
              new MapPartitionsRDD(
                this,
                (context: TaskContext, index: Int, iter: Iterator[T]) => cleanedF(iter),
                preservesPartitioning)
            }
     */

    val func1 = (e: Int) => e * 10
    val mapPartitionsRdd: MapPartitionsRDD[Int, Int] =
      new MapPartitionsRDD[Int, Int](dataRDD, (_, _, _iterator) => _iterator.map(func1))

    println("map(): ", mapRDD.collect().toBuffer)
    println("MapPartitionsRDD--map: ", mapPartitionsRdd.collect().toBuffer)

    /*--------------------------------------------------------------*/
    val filterRDD: RDD[Int] = dataRDD.filter(_ % 2 == 0)

    val func2 = (e: Int) => e % 2 == 0
    val mapPartitionsRDDFilter: MapPartitionsRDD[Int, Int] =
      new MapPartitionsRDD[Int, Int](dataRDD, (_, _, _iterator) => _iterator.filter(func2))

    println("filterRDD: ", filterRDD.collect().toBuffer)
    println("MapPartitionsRDD--filter: ", mapPartitionsRDDFilter.collect().toBuffer)

    sc.stop()
  }

}
